package org.apache.flink.peter.kafka;

import java.util.Properties;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer010;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer09;
import org.apache.flink.util.Collector;

/**
 * 将字符串上传到Kafka中，Flink从Kafka中读取字符串。
 * @author peter
 *
 */
@SuppressWarnings("serial")
public class KafkaWordCount {
    public static void main(String[] args) throws Exception {
        Properties properties = new Properties();
        properties.put("bootstrap.servers", "localhost:9092");
        properties.setProperty("group.id", "test");
        FlinkKafkaConsumer09<String> consumer = new FlinkKafkaConsumer09<String>("test", new SimpleStringSchema(), properties);
        consumer.setStartFromEarliest();

        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStream<String> stream = env.addSource(consumer);
        stream.flatMap(new FlatMapFunction<String, WordWithCount>() {
            @Override
            public void flatMap(String value, Collector<WordWithCount> out) throws Exception {
                for(String word: value.split(" ")) {
                    if(null != word && !"".equals(word)) {
                        out.collect(new WordWithCount(word, 1));
                    }
                }
            }
        })
        .keyBy("word")
        .timeWindow(Time.seconds(5))
        .reduce(new ReduceFunction<WordWithCount>() {
            @Override
            public WordWithCount reduce(WordWithCount value1, WordWithCount value2)
                    throws Exception {
                return new WordWithCount(value1.word, value1.count + value2.count);
            }
        })
        .print().setParallelism(1);
        env.execute();
    }

    /**
     * Data type for words with count.
     */
    public static class WordWithCount {
        public String word;
        public long count;

        public WordWithCount() {}

        public WordWithCount(String word, long count) {
            this.word = word;
            this.count = count;
        }

        @Override
        public String toString() {
            return word + " : " + count;
        }
    }
}
